package cmd

import (
	"context"
	"devops-super/internal/consts"
	"devops-super/internal/model/mymodel"
	"devops-super/internal/queues"
	"devops-super/internal/service"
	"devops-super/utility/util/appUtil"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gcmd"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
	"time"
)

var (
	MqttSub = gcmd.Command{
		Name:  "MqttSub",
		Usage: "MqttSub",
		Brief: "start MqttSub",
		Func: func(ctx context.Context, parser *gcmd.Parser) (err error) {
			SetTimezone(ctx)
			g.Log().Info(ctx, "MqttSub start ..")
			register()
			startConnectMqtt(ctx)

			return err
		},
	}
)

func startQueue(ctx context.Context) {
	// 创建队列
	que1 := queues.GetOrCreateQueue(consts.QUEUE_AE_PARAM, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	que2 := queues.GetOrCreateQueue(consts.QUEUE_AE_WAVE, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	que3 := queues.GetOrCreateQueue(consts.QUEUE_VI_PARAM, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	que4 := queues.GetOrCreateQueue(consts.QUEUE_VI_WAVE, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	que5 := queues.GetOrCreateQueue(consts.QUEUE_AE_VI_PARAM, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	que6 := queues.GetOrCreateQueue(consts.QUEUE_AE_VI_WAVE, queues.QueueConfig{
		MaxQueueSize: 10000,
		BatchSize:    1000,
		MaxInterval:  5 * time.Second,
	}, func(i []interface{}) {

	})
	// 启动队列
	go que1.Start()
	go que2.Start()
	go que3.Start()
	go que4.Start()
	go que5.Start()
	go que6.Start()
}

func startConnectMqtt(ctx context.Context) {
	done := make(chan bool)
	confKey := consts.EnvConfig_Mqtt
	confString, err := g.Cfg().Get(ctx, confKey)
	if err != nil || confString == nil {
		glog.Line().Info(ctx, "read mqtt config empty")
		return
	}
	var confs map[string]*mymodel.MqttConnectOption
	err = gconv.Structs(confString, &confs)
	if err != nil || confs == nil {
		glog.Line().Info(ctx, "parse mqtt config empty")
		return
	}
	topics := []string{
		consts.EMQX_TOPIC_PUB + "#",
		consts.EMQX_TOPIC_WILL + "#",
		consts.EMQX_TOPIC_AppPub + "#",
		consts.EMQX_TOPIC_DEVICE_ON,
		consts.EMQX_TOPIC_DEVICE_OFF,
		consts.MOCHI_MQTT_DEVICE_ON + "#",
		consts.MOCHI_MQTT_DEVICE_OFF + "#",
	}
	if confs != nil {
		for _, mqttConf := range confs {
			go func(mqttConf *mymodel.MqttConnectOption) {
				ClientId := appUtil.GenerateMqttClientId("s")
				mqttConf.ClientId = ClientId
				mqttConf.Topics = topics
				//开启mqtt连接
				err := service.Aeemqx().DoEmqxSub(ctx, mqttConf)
				if err != nil {
					g.Log().Info(ctx, "MqttSub err:", err.Error())
					return
				}
			}(mqttConf)
		}
	}
	<-done
}
